package com.intct.flink;

import com.intct.hbase.bean.Book;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * @author gufg
 * @since 2025-06-26 15:48
 */
public class MySQLJDBCDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.fromElements(
                new Book(1, "老人与海", "张三", 56.66, 10),
                new Book(2, "活着", "李四", 76.66, 20),
                new Book(3, "飞鸟", "王五", 56.66, 5)
                ).addSink(JdbcSink.sink(
                "insert into book (id, title, author, price, qty) values (?,?,?,?,?)",
                new JdbcStatementBuilder<Book>() {
                    @Override
                    public void accept(PreparedStatement ps, Book book) throws SQLException {
                        ps.setInt(1, book.getId());
                        ps.setString(2, book.getTitle());
                        ps.setString(3, book.getAuthor());
                        ps.setDouble(4, book.getPrice());
                        ps.setInt(5, book.getQty());
                    }
                },
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withUrl("jdbc:mysql://cdh-node:3306/travel?serverTimezone=Asia/Shanghai&useUnicode=true&characterEncoding=UTF-8")
                        .withUsername("root")
                        .withPassword("999999")
                        .withDriverName("com.mysql.cj.jdbc.Driver")
                        .build()));
        env.execute();
    }
}
